package com.atguigu.chapter13;

import com.atguigu.Bean.UserBehavior;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.hash.BloomFilter;
import org.apache.flink.shaded.guava18.com.google.common.hash.Funnels;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @ClassName: Flink02_BloomFilter
 * @Description:
 * @Author: kele
 * @Date: 2021/4/15 11:34
 *
 * 统计一小时内的登录用户数
 *
 *
 **/
public class Flink02_BloomFilter {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建WatermarkStrategy
        WatermarkStrategy<UserBehavior> wms = WatermarkStrategy
                .<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                    @Override
                    public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                        return element.getTimestamp() * 1000L;
                    }
                });

        env
                .readTextFile("in/UserBehavior.csv")
                .map(line -> { // 对数据切割, 然后封装到POJO中
                    String[] split = line.split(",");
                    return new UserBehavior(Long.valueOf(split[0]), Long.valueOf(split[1]), Integer.valueOf(split[2]), split[3], Long.valueOf(split[4]));
                })
                .filter(behavior -> "pv".equals(behavior.getBehavior())) //过滤出pv行为
                .assignTimestampsAndWatermarks(wms)
                .keyBy(UserBehavior::getBehavior)
                .window(TumblingEventTimeWindows.of(Time.minutes(60)))

                .process(new ProcessWindowFunction<UserBehavior,String, String, TimeWindow>() {

                    private ValueState<Long> numberState;
                    private ValueState<BloomFilter<Long>> userState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        //定义一个value状态，保存bloomFikter
                        userState = getRuntimeContext()
                                .getState(new ValueStateDescriptor<BloomFilter<Long>>("userState",
                                        TypeInformation.of(new TypeHint<BloomFilter<Long>>() {
                                        })
                                ));

                        numberState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("numberState", Long.class));


                    }

                    @Override
                    public void process(String s,
                                        Context context,
                                        Iterable<UserBehavior> elements,
                                        Collector<String> out) throws Exception {

                        if(userState.value() == null){
                            /**
                             *
                             * 参数1：
                             * 参数2：期望插入数据的数目
                             * 参数3：错误率
                             */
                            BloomFilter<Long> bloomFilter = BloomFilter.create(Funnels.longFunnel(), 100000, 0.01);

                            userState.update(bloomFilter);
                            numberState.update(0l);

                        }

                        BloomFilter<Long> bf = userState.value();
                        Long num = numberState.value();

                        for (UserBehavior element : elements) {

                            if( bf.put(element.getUserId())){
                                num = num+1;
                                numberState.update(num);
                            }
                        }


                        out.collect(numberState.value()+"    "+context.window());

                        userState.clear();


                    }
                })
                .print();

        env.execute();

    }

}
